﻿using KafkaSDK;

namespace DDDDemoABP1.Payments;

public class KafkaAuditMessageSenderProvider : IAuditMessageSenderProvider
{
    private readonly KafkaTemplate<string, string> _KafkaTemplate;

    public const string ProviderName = "Kafka";

    private const string TOPIC_AUDIT_LOG = "topic_audit_log";

    public string Name => ProviderName;

    public KafkaAuditMessageSenderProvider(KafkaTemplate<string, string> kafkaTemplate)
    {
        _KafkaTemplate = kafkaTemplate;
    }

    public async Task<TransferAuditMessageSendStatus> SendAsync(TransferAuditMessage message)
    {
        var messageText = ParseToMessageText(message);
        await _KafkaTemplate.SendAsync(TOPIC_AUDIT_LOG, messageText);
        return TransferAuditMessageSendStatus.Success;
    }

    private static string ParseToMessageText(TransferAuditMessage message)
    {
        return $"Kafka message:{message.UserId},{message.MerchantAccountNo},{message.Amount}";
    }
}
